package com.lgj.demo.query;

import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;

/**
 * @Author
 * @ClassName MessageQuery
 * @Date 2022/11/18 15:35
 * @Description 消息队列
 * @Version 1.0
 */
@Slf4j
@Component
public class MessageQuery{

    private static final String REDIS_LP_QUEUE = "redis:lp:queue";
    private static final String REDIS_RP_QUEUE = "redis:rp:queue";

    private static ThreadPoolExecutor lpThreadPoolExecutor;
    private static ThreadPoolExecutor rpThreadPoolExecutor;

    static {
        lpThreadPoolExecutor = new ThreadPoolExecutor(16, 16, 600L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(65536), r -> {
            Thread t = Executors.defaultThreadFactory().newThread(r);
            t.setDaemon(true);
            return t;
        });
        rpThreadPoolExecutor = new ThreadPoolExecutor(16,16,600L, TimeUnit.SECONDS,new ArrayBlockingQueue<>(65536), r -> {
            Thread t = Executors.defaultThreadFactory().newThread(r);
            t.setDaemon(true);
            return t;
        });
    }

    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    RedisTemplate redisTemplate;

    @PostConstruct
    public void rightMonitor() {
        log.info("rightMonitor ===>>>");
        rpThreadPoolExecutor.execute(()->{
            // 获取消息数量
            while (true){
                Object o = stringRedisTemplate.opsForList().rightPop(REDIS_RP_QUEUE, 0, TimeUnit.SECONDS);
                log.info("先进后出队列 接收到数据：{}", o);
                try {
                    // 防止CPU频繁空转
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        log.info("rpThreadPoolExecutor end ===>>>");
    }

    @PostConstruct
    public void leftMonitor() {
        log.info("leftMonitor ===>>>");
        lpThreadPoolExecutor.execute(()->{
            while (true){
                Object o = stringRedisTemplate.opsForList().leftPop(REDIS_LP_QUEUE, 0, TimeUnit.SECONDS);
                log.info("先进先出队列 接收到数据：{}", o);
                try {
                    // 防止CPU频繁空转
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        log.info("lpThreadPoolExecutor end ===>>>");
    }


    /**
     * 获取对应key左侧第一个元素并删除
     * @param key
     * @return
     */
    public Object leftPop(String key){
        return stringRedisTemplate.opsForList().leftPop(key);
    }

    /**
     * 右侧插入集合中
     * @param key
     * @param value
     */
    public void rightPush(String key, String value){
        stringRedisTemplate.opsForList().rightPush(key, value);
    }
}
